package com.learn.lb.mapred;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * @author laibo
 * @since 2019/7/1 17:26
 * ===================================
 * <p> 数据去重
 * <p> 因为key是唯一的，
 * <p> 所以把每一行数据当成一个key，reduce的时候直接输出key就可以了
 * ===================================
 */
public class DistinctData {

    private static Logger logger = LoggerFactory.getLogger(DistinctData.class);

    /**
     * map去重
     * KEYIN(K1)： 第一个，分片之后的Key，Map运算接受的Key类型
     * VALUEIN(V1): 第二个 表示行文本内容，Map运行接收的Value类型
     * KEYOUT(K2): 第三个 Map运算之后输出的Key
     * VALUEOUT(V2)：第四个： Map运行之后输出的Value
     */
    public static class Map extends Mapper<Object, Text, Text, Text> {

        /**
         * @param key   K1
         * @param value V1
         *              <p> 输入 <K1, V1> 输出 <K2, V2>
         */
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            //value是一行数据，直接把该行数据当成key输出到reduce中
            logger.info("map读取到的数据：key:{}, value:{}", key, value);
            //输出<K2, V2>
            context.write(value, new Text(""));
        }
    }

    /**
     * reduce将输入中的key复制到输出数据的key上,并直接输出
     * KEYIN(K1)： 第一个，Map之后的Key类型
     * VALUEIN(V1): 第二个 Map之后的Value类型
     * KEYOUT(K2): 第三个，reducer之后的Key类型
     * VALUEOUT(V2)：第四个，reducer之后的Value类型
     */
    public static class Reduce extends Reducer<Text, Text, Text, Text> {

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            //输入<K2, {V2....}> 输出 <K2>
            logger.info("reduce，key:{}", key);
            context.write(key, new Text(""));
        }
    }

    public static void main(String[] args) throws Exception {
        logger.info("开始运行mapreducer任务.....");
        Configuration hdfsConfiguration = new Configuration();
        Path dstPath = new Path("/distinct_in/data");
        Path outPath = new Path("/distinct_out");
        GenerateMapReducerUtils.buildJobBefore(hdfsConfiguration, "distinct_data.txt", dstPath, outPath);
        logger.info("开始进行MapReducer计算....");
        Job job = Job.getInstance(hdfsConfiguration, "Data Deduplication");
        job.setJarByClass(DistinctData.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        //设置输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        //设置输出目录
        FileInputFormat.addInputPath(job, dstPath);
        FileOutputFormat.setOutputPath(job, outPath);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
